package com.example.flinkcourse.lesson5.source;

import com.example.flinkcourse.lesson5.config.Config;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * MySQL CDC Source 实现
 */
public class MySqlCdcSource {
    private static final Logger LOG = LoggerFactory.getLogger(MySqlCdcSource.class);

    /**
     * 创建 MySQL CDC Source
     *
     * @param env Flink 执行环境
     * @param config 配置信息
     * @return DataStreamSource
     */
    public static DataStreamSource<String> createSource(
            StreamExecutionEnvironment env,
            Config config) {
        LOG.info("创建 MySQL CDC Source...");
        
        // 创建 MySQL CDC Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname(config.getMysqlHost())
            .port(config.getMysqlPort())
            .databaseList(config.getMysqlDatabase())
            .tableList(config.getMysqlDatabase() + "\\..*")  // 同步所有表
            .username(config.getMysqlUsername())
            .password(config.getMysqlPassword())
            .deserializer(new JsonDebeziumDeserializationSchema())  // 使用 JSON 格式
            .build();
        
        // 创建数据流
        return env.fromSource(
            mySqlSource,
            WatermarkStrategy.noWatermarks(),
            "MySQL CDC Source"
        );
    }
} 